package 数据预处理

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{DoubleType, IntegerType, StringType}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author 35192
 * @date 2021-05-21 16:57
 */
object Task02 {
  def main(args: Array[String]): Unit = {
    // 创建环境
    val conf = new SparkConf()
      .setAppName("log_analysis")
      .setMaster("local[6]")
    val sc = new SparkContext(conf)
    val spark = SparkSession.builder()
      .appName("task")
      .master("local[6]")
      .getOrCreate()
    // 导入隐式转换和函数
    import org.apache.spark.sql.functions._
    import spark.implicits._

    // 读取数据集 --- RDD
    val data = sc.textFile("G:\\Projects\\IdeaProjects\\Spark_Competition\\dataset\\test")
      .map {
        line =>
          var items = line.split(",")
          (items(0), items(1), items(2), items(3), items(4), items(5))
      }
    data.foreach(println(_))


    val missing = data.map{
      row =>
        var counts = 0
        if (row._1.equals("None")){
          counts = counts + 1;
        }
        if (row._2.equals("None")){
          counts = counts + 1;
        }
        if (row._3.equals("None")){
          counts = counts + 1;
        }
        if (row._4.equals("None")){
          counts = counts + 1;
        }
        if (row._5.equals("None")){
          counts = counts + 1;
        }
        if(row._6.equals("None")){
          counts = counts + 1;
        }
        (row._1,counts)
    }
    missing.foreach(println(_))

    // 转为DF
    val df = data.toDF("id", "weight", "height", "age", "gender", "income")
    df.show()

    val df0 = df
      .select('id, 'weight, 'height, 'age, 'gender, 'income)
      .where('income === "None")
    df0.show()

    val df1 = df
      .select('id cast (IntegerType), 'weight cast (DoubleType), 'height cast (DoubleType), 'age cast (IntegerType), 'gender , 'income cast (IntegerType))
      .where('id === 3)
    df1.show()

    val df2 = df
      .select('id cast (IntegerType), 'weight cast (DoubleType), 'height cast (DoubleType), 'age cast (IntegerType), 'gender cast (StringType), 'income cast (IntegerType))
      .agg(
        count('id === null),
        count('weight === null)
//        count('height === null),
//        count('age === null) ,
//        count('gender === "None") ,
//        count('income === null)
      )

    df2.show()
  }
}


